package com.usoft.example.search;

import com.usoft.grpc.example.search.SearchRequest;
import com.usoft.grpc.example.search.SearchResponse;
import com.usoft.grpc.example.search.SearchServiceGrpc;
import io.grpc.stub.StreamObserver;

import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;

/**
 * 服务器业务逻辑实现
 * Created by xinxingegeya on 15/9/17.
 */
public class SearchServiceImpl implements SearchServiceGrpc.SearchService {

    private static final Logger logger = Logger
        .getLogger(SearchServiceImpl.class.getName());

    /**
     * Simple RPC
     * A simple RPC where the client sends a request to the server using the
     * stub and waits for a response to come back, just like a normal function
     * call.
     * 
     * @param request
     * @param responseObserver
     */
    @Override
    public void searchWithSimpleRpc(SearchRequest request,
            StreamObserver<SearchResponse> responseObserver) {
        System.out.println("pageNo=" + request.getPageNumber());
        System.out.println("query string=" + request.getQuery());
        System.out.println("pageSize=" + request.getResultPerPage());

        List<SearchResponse.Result> results = new ArrayList<SearchResponse.Result>(
            10);

        for (int i = 0; i < request.getResultPerPage(); i++) {
            SearchResponse.Result result = SearchResponse.Result.newBuilder()
                .setTitle("title" + i).setUrl("dev.usoft.com")
                .addSnippets("snippets" + i).build();
            results.add(result);
        }
        SearchResponse response = SearchResponse.newBuilder()
            .addAllResult(results).build();
        responseObserver.onNext(response);
        //We use the response observer's onCompleted() method to specify that we've finished dealing with the RPC.
        responseObserver.onCompleted();
    }

    /**
     * Server-side streaming RPC
     * A server-side streaming RPC where the client sends a request to the
     * server and gets a stream to read a sequence of messages back. The client
     * reads from the returned stream until there are no more messages. As you
     * can see in our example, you specify a server-side streaming method by
     * placing the stream keyword before the response type.
     * 
     * @param request
     * @param responseObserver
     */
    @Override
    public void searchWithServerSideStreamRpc(SearchRequest request,
            StreamObserver<SearchResponse> responseObserver) {

        System.out.println("pageNo=" + request.getPageNumber());
        System.out.println("query string=" + request.getQuery());
        System.out.println("pageSize=" + request.getResultPerPage());

        List<SearchResponse.Result> results = new ArrayList<SearchResponse.Result>(
            10);

        for (int i = 0; i < request.getResultPerPage(); i++) {
            SearchResponse.Result result = SearchResponse.Result.newBuilder()
                .setTitle("title" + i).setUrl("dev.usoft.com")
                .addSnippets("snippets" + i).build();
            results.add(result);
        }
        SearchResponse response = SearchResponse.newBuilder()
            .addAllResult(results).build();
        responseObserver.onNext(response);

        SearchResponse.Result result = SearchResponse.Result.newBuilder()
            .setTitle("title").setUrl("dev.usoft.com").addSnippets("the last")
            .build();
        SearchResponse theNext = SearchResponse.newBuilder().addResult(result)
            .build();
        responseObserver.onNext(theNext);
        responseObserver.onCompleted();
    }

    /**
     * Client-side streaming RPC
     * A client-side streaming RPC where the client writes a sequence of
     * messages and sends them to the server, again using a provided stream.
     * Once the client has finished writing the messages, it waits for the
     * server to read them all and return its response. You specify a
     * server-side streaming method by placing the stream keyword before the
     * request type.
     * 
     * @param responseObserver
     * @return
     */
    @Override
    public StreamObserver<SearchRequest> searchWithClientSideStreamRpc(
            final StreamObserver<SearchResponse> responseObserver) {
        return new StreamObserver<SearchRequest>() {
            int searchCount;
            SearchRequest previous;
            long startTime = System.nanoTime();

            @Override
            public void onNext(SearchRequest searchRequest) {
                searchCount++;
                if (previous != null
                    && previous.getResultPerPage() == searchRequest
                        .getResultPerPage()
                    && previous.getPageNumber() == searchRequest
                        .getPageNumber()) {
                    logger.info("do nothing");
                    return;
                }
                previous = searchRequest;
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("error");
            }

            @Override
            public void onCompleted() {
                logger.info("search count = " + searchCount);
                List<SearchResponse.Result> results = new ArrayList<SearchResponse.Result>(
                    10);
                for (int i = 0; i < previous.getResultPerPage(); i++) {
                    SearchResponse.Result result = SearchResponse.Result
                        .newBuilder().setTitle("title" + i)
                        .setUrl("dev.usoft.com").addSnippets("snippets" + i)
                        .build();
                    results.add(result);
                }
                SearchResponse response = SearchResponse.newBuilder()
                    .addAllResult(results).build();
                responseObserver.onNext(response);
                responseObserver.onCompleted();
                logger.info("spend time = "
                    + String.valueOf(System.nanoTime() - startTime));
            }
        };
    }

    /**
     * Bidirectional streaming RPC
     * A bidirectional(双向的) streaming RPC where both sides send a sequence of
     * messages using a read-write stream. The two streams operate
     * independently, so clients and servers can read and write in whatever
     * order they like: for example, the server could wait to receive all the
     * client messages before writing its responses, or it could alternately
     * read a message then write a message, or some other combination of reads
     * and writes. The order of messages in each stream is preserved. You
     * specify this type of method by placing the stream keyword before both the
     * request and the response.
     * 
     * @param responseObserver
     * @return
     */
    @Override
    public StreamObserver<SearchRequest> searchWithBidirectionalStreamRpc(
            final StreamObserver<SearchResponse> responseObserver) {
        return new StreamObserver<SearchRequest>() {
            int searchCount;
            SearchRequest previous;
            long startTime = System.nanoTime();

            @Override
            public void onNext(SearchRequest searchRequest) {
                searchCount++;
                if (previous != null
                    && previous.getResultPerPage() == searchRequest
                        .getResultPerPage()
                    && previous.getPageNumber() == searchRequest
                        .getPageNumber()) {
                    logger.info("do nothing");
                    return;
                }
                previous = searchRequest;

                logger.info("search count = " + searchCount);
                List<SearchResponse.Result> results = new ArrayList<SearchResponse.Result>(
                    10);
                for (int i = 0; i < searchRequest.getResultPerPage(); i++) {
                    SearchResponse.Result result = SearchResponse.Result
                        .newBuilder().setTitle("title" + i)
                        .setUrl("dev.usoft.com").addSnippets("snippets" + i)
                        .build();
                    results.add(result);
                }
                SearchResponse response = SearchResponse.newBuilder()
                    .addAllResult(results).build();
                responseObserver.onNext(response);
                logger.info("spend time = "
                    + String.valueOf(System.nanoTime() - startTime));
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("error");
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }

}
